home *** CD-ROM | disk | FTP | other *** search
/ Personal Computer World 2008 February / PCWFEB08.iso / Software / Freeware / Miro 1.0 / Miro_Installer.exe / Miro_Downloader.exe / test / schedulertest.pyc (.txt) < prev    next >
Encoding:
Python Compiled Bytecode  |  2007-11-12  |  3.0 KB  |  88 lines
  1. # Source Generated with Decompyle++
  2. # File: in.pyc (Python 2.5)
  3.  
  4. import unittest
  5. from time import time, sleep
  6. import threading
  7. import eventloop
  8. from test.framework import EventLoopTest
  9.  
  10. class SchedulerTest(EventLoopTest):
  11.     
  12.     def setUp(self):
  13.         self.gotArgs = []
  14.         self.gotKwargs = []
  15.         EventLoopTest.setUp(self)
  16.  
  17.     
  18.     def callback(self, *args, **kwargs):
  19.         self.gotArgs.append(args)
  20.         self.gotKwargs.append(kwargs)
  21.         if 'stop' in kwargs.keys():
  22.             eventloop.quit()
  23.         
  24.  
  25.     
  26.     def testCallbacks(self):
  27.         eventloop.addIdle(self.callback, 'foo')
  28.         eventloop.addTimeout(0.1, self.callback, 'foo', args = ('chris',), kwargs = {
  29.             'hula': 'hula' })
  30.         eventloop.addTimeout(0.2, self.callback, 'foo', args = ('ben',), kwargs = {
  31.             'hula': 'moreHula',
  32.             'stop': 1 })
  33.         self.runEventLoop()
  34.         self.assertEquals(self.gotArgs[0], ())
  35.         self.assertEquals(self.gotArgs[1], ('chris',))
  36.         self.assertEquals(self.gotArgs[2], ('ben',))
  37.         self.assertEquals(self.gotKwargs[0], { })
  38.         self.assertEquals(self.gotKwargs[1], {
  39.             'hula': 'hula' })
  40.         self.assertEquals(self.gotKwargs[2], {
  41.             'hula': 'moreHula',
  42.             'stop': 1 })
  43.  
  44.     
  45.     def testQuitWithStuffStillScheduled(self):
  46.         eventloop.addTimeout(0.1, self.callback, 'foo', kwargs = {
  47.             'stop': 1 })
  48.         eventloop.addTimeout(2, self.callback, 'foo')
  49.         self.runEventLoop()
  50.         self.assertEquals(len(self.gotArgs), 1)
  51.  
  52.     
  53.     def testTiming(self):
  54.         startTime = time()
  55.         eventloop.addTimeout(0.2, self.callback, 'foo', kwargs = {
  56.             'stop': 1 })
  57.         self.runEventLoop()
  58.         endTime = time()
  59.         self.assertAlmostEqual(startTime + 0.2, endTime, places = 1)
  60.  
  61.     
  62.     def testLotsOfThreads(self):
  63.         timeouts = [
  64.             0,
  65.             0,
  66.             0.1,
  67.             0.2,
  68.             0.3]
  69.         threadCount = 8
  70.         
  71.         def thread():
  72.             sleep(0.5)
  73.             for timeout in timeouts:
  74.                 eventloop.addTimeout(timeout, self.callback, 'foo')
  75.             
  76.  
  77.         for i in range(threadCount):
  78.             t = threading.Thread(target = thread)
  79.             t.start()
  80.         
  81.         eventloop.addTimeout(1, self.callback, 'foo', kwargs = {
  82.             'stop': 1 })
  83.         self.runEventLoop()
  84.         totalCalls = len(timeouts) * threadCount + 1
  85.         self.assertEquals(len(self.gotArgs), totalCalls)
  86.  
  87.  
  88.